package com.example.concurrency.threadpool.completableFuture;

import com.example.common.util.TimeUtil;
import com.example.concurrency.threadpool.MyRunnable;
import com.example.concurrency.threadpool.MyThreadPoolExecutor;

import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.concurrent.*;
import java.util.function.BiConsumer;
import java.util.function.BiFunction;
import java.util.function.Function;
import java.util.function.Supplier;

public class CompletableFutureDemo
{
    private ExecutorService threadPool = new MyThreadPoolExecutor(5, 10, 1,
            TimeUnit.SECONDS,
            new LinkedBlockingDeque<>(5),
            new ThreadPoolExecutor.AbortPolicy());

    public static void main(String[] args)
    {
        CompletableFutureDemo demo = new CompletableFutureDemo();

//        demo.testSubmit();
//        demo.testThenCombine();
//        demo.testThenApplyCombine();
//        demo.testAllofMaximumPoolSize();
//        demo.testAllofRejected();
//        demo.testMoreThreadsCompletableFutureGet();
//        demo.testAllOfException();
//        demo.testMainCompletableFuture();
//        demo.testCompletedFuture();
        demo.testWhenComplete();

//        try
//        {
//            ConcurrentHashMap<Date[], Integer> map = new ConcurrentHashMap<>();
//            Date[] dates1 = new Date[]{TimeUtil.getDateFromTimeString("2019-12-05 00:01:23"), TimeUtil.getDateFromTimeString("2019-12-06 00:01:23")};
//            Date[] dates2 = new Date[]{TimeUtil.getDateFromTimeString("2019-12-05 00:01:23"), TimeUtil.getDateFromTimeString("2019-12-06 00:01:23")};
//            map.put(dates1, 1);
//            System.out.println(map.get(dates1));
//        } catch (Exception e)
//        {
//            e.printStackTrace();
//        }

    }

    private void testSubmit()
    {
        System.out.println("start");
        for (int i = 0; i < 15; i++)
        {
            CompletableFuture<Integer> future = CompletableFuture.supplyAsync(new MySupplier(i), threadPool);
            future.thenRun(() -> System.out.println(Thread.currentThread().getName() + "-----------"));
        }
        System.out.println("end");
    }

    private void testThenCombine()
    {
        System.out.println("start");
        CompletableFuture<Integer> future5 = CompletableFuture.supplyAsync(new MySupplier(5), threadPool);
        CompletableFuture<Integer> future6 = CompletableFuture.supplyAsync(new MySupplier(6), threadPool);
        try
        {
            future5.thenCombine(future6, Integer::sum).get();
        } catch (InterruptedException | ExecutionException e)
        {
            e.printStackTrace();
        }
        System.out.println("end");
    }

    private void testThenApplyCombine()
    {
        System.out.println("start");
        Function<Integer, Integer> function = integer ->
        {
            System.out.println("function: " + 0);
            return 0;
        };
        CompletableFuture<Integer> future5 = CompletableFuture.supplyAsync(new MySupplier(5), threadPool);
        CompletableFuture<Integer> future6 = CompletableFuture.supplyAsync(new MySupplier(10), threadPool);
        try
        {
            future5.thenApply(function).thenCombine(future6, Integer::sum).get();
        } catch (InterruptedException | ExecutionException e)
        {
            e.printStackTrace();
        }
        System.out.println("end");
    }

    /**
     * 最大线程数
     * max(corePoolSize,maximumPoolSize) + 阻塞队列长度
     */
    private void testAllofMaximumPoolSize()
    {
        System.out.println("start");
        List<CompletableFuture> completableFutureList = new ArrayList<>();
        for (int i = 0; i < 15; i++)
        {
            CompletableFuture<Integer> future = CompletableFuture
                    .supplyAsync(new MySupplier(i), threadPool)
                    .thenApply(integer ->
                    {
                        System.out.println("---------------------" + integer);
                        return integer;
                    });
            completableFutureList.add(future);
        }
        CompletableFuture completableFuture = CompletableFuture
                .allOf(completableFutureList.toArray(new CompletableFuture[0]));
        completableFuture.thenRun(() -> System.out.println("==========================="));
//        try
//        {
//            completableFuture.get();
//        } catch (InterruptedException | ExecutionException e)
//        {
//            e.printStackTrace();
//        }
        System.out.println("end");
    }

    /**
     * 多个线程访问同一个执行中的CompletableFuture.get()
     * fan-in/fan-out
     */
    private void testMoreThreadsCompletableFutureGet()
    {
        List<CompletableFuture> completableFutureList = new ArrayList<>();
        for (int i = 0; i < 15; i++)
        {
            CompletableFuture<Integer> future = CompletableFuture.supplyAsync(new MySupplier(i), threadPool);
            completableFutureList.add(future);
        }
        CompletableFuture completableFuture = CompletableFuture
                .allOf(completableFutureList.toArray(new CompletableFuture[0]));
        for (int i = 0; i < 5; i++)
        {
            new Thread("thread " + i)
            {
                @Override
                public void run()
                {
                    try
                    {
                        System.out.println(Thread.currentThread().getName() + "start");
                        completableFuture.get();
                    } catch (InterruptedException | ExecutionException e)
                    {
                        e.printStackTrace();
                    } finally
                    {
                        System.out.println(Thread.currentThread().getName() + "end");
                    }
                }
            }.start();
        }
    }

    /**
     * 嵌套CompletableFuture
     * 符合fan-in/fan-out
     */
    private void testMainCompletableFuture()
    {
        CompletableFuture completableFuture = CompletableFuture.supplyAsync(() ->
        {
            System.out.println("main completableFuture start");
            List<CompletableFuture> list = createCompletableFutures(5);
            try
            {
                CompletableFuture.allOf(list.toArray(new CompletableFuture[0])).get();
                throw new RuntimeException("z z z z ");
            } catch (Exception e)
            {
                System.out.println(e);
            }
            System.out.println("main completableFuture end");
            return 0;
        });

        try
        {
            TimeUnit.SECONDS.sleep(5);
        } catch (InterruptedException e)
        {
            e.printStackTrace();
        }
        System.out.println("-------------------------------");

        for (int i = 0; i < 5; i++)
        {
            new Thread("thread " + i)
            {
                @Override
                public void run()
                {
                    try
                    {
                        System.out.println(Thread.currentThread().getName() + "start");
                        completableFuture.get();
                    } catch (InterruptedException | ExecutionException e)
                    {
                        e.printStackTrace();
                    } finally
                    {
                        System.out.println(Thread.currentThread().getName() + "end");
                    }
                }
            }.start();
        }
    }

    private List<CompletableFuture> createCompletableFutures(int count)
    {
        List<CompletableFuture> completableFutureList = new ArrayList<>();
        for (int i = 0; i < count; i++)
        {
            CompletableFuture<Integer> future = CompletableFuture.supplyAsync(new MySupplier(i), threadPool);
            completableFutureList.add(future);
        }
        return completableFutureList;
    }

    /**
     * 如果线程数量超过执行的线程池的最大线程数，如果future被拒绝了，CompletableFuture.get()会一致阻塞
     */
    private void testAllofRejected()
    {
        System.out.println("start");
        List<CompletableFuture> completableFutureList = new ArrayList<>();
        for (int i = 0; i < 16; i++)
        {
            CompletableFuture<Integer> future = CompletableFuture.supplyAsync(new MySupplier(i), threadPool);
            completableFutureList.add(future);
        }
        CompletableFuture completableFuture = CompletableFuture
                .allOf(completableFutureList.toArray(new CompletableFuture[0]));
        try
        {
            completableFuture.get();
        } catch (InterruptedException | ExecutionException e)
        {
            e.printStackTrace();
        }
        System.out.println("end");
    }

    private void testCompletedFuture()
    {
        CompletableFuture future = CompletableFuture.completedFuture("1234").thenApply(s ->
        {
            System.out.println("========================= " + s);
            return s;
        });
        try
        {
            TimeUnit.SECONDS.sleep(3);
        } catch (InterruptedException e)
        {
            e.printStackTrace();
        }
        System.out.println("-----------------------------------------------");

        try
        {
            future.get();
        } catch (InterruptedException | ExecutionException e)
        {
            e.printStackTrace();
        }
    }

    /**
     * 如果线程执行过程中发生异常，CompletableFuture.get()能正常结束
     * 并且最后会返回异常
     */
    private void testAllOfException()
    {
        System.out.println("start");
        CompletableFuture<Integer> exceptionFuture = CompletableFuture.supplyAsync(() ->
        {
            try
            {
                Thread.sleep(3000);
            } catch (InterruptedException e)
            {
                e.printStackTrace();
            }
            throw new RuntimeException("4 异常");
        }, threadPool);
        List<CompletableFuture> completableFutureList = new ArrayList<>();
        for (int i = 0; i < 14; i++)
        {
            CompletableFuture<Integer> future = CompletableFuture.supplyAsync(new MySupplier(i), threadPool);
            completableFutureList.add(future);
        }
        completableFutureList.add(4, exceptionFuture);
        CompletableFuture completableFuture = CompletableFuture
                .allOf(completableFutureList.toArray(new CompletableFuture[0]));
        try
        {
            completableFuture.get();
        } catch (InterruptedException | ExecutionException e)
        {
            e.printStackTrace();
        }
        System.out.println("end");
    }

    private void testWhenComplete()
    {
        CompletableFuture future = CompletableFuture.supplyAsync(() -> "1", threadPool)
                .thenApply(s -> "1" + s)
                .thenApply(s ->
                {
                    System.out.println("11111111111111111111111111");
                    try
                    {
                        TimeUnit.SECONDS.sleep(1);
                    } catch (InterruptedException e)
                    {
                        e.printStackTrace();
                    }
                    System.out.println("aaaaaaaaaaaaaaaaaaaaaaaaaaaa");
//                    return s + "1";
                    throw new RuntimeException("-------------");
                })
                .whenComplete((o, throwable) -> System.out.println("whenComplete " + (o != null ? o : "") + (throwable == null ? "" : throwable.getMessage())))
                .whenComplete((o, throwable) -> System.out.println("whenComplete============ " + (o != null ? o : "") + (throwable == null ? "" : throwable.getMessage())));
        System.out.println("222222222222222222222222");
        try
        {
            TimeUnit.SECONDS.sleep(3);
        } catch (InterruptedException e)
        {
            e.printStackTrace();
        }
        System.out.println("bbbbbbbbbbbbbbbbbbbbbbbbbbbbbb");
        System.out.println("is done: " + future.isDone());
        System.out.println("is complete: " + future.isCompletedExceptionally());
    }

    public static class MySupplier implements Supplier<Integer>, MyRunnable
    {
        private final Integer num;

        private String name;

        private MySupplier(Integer num)
        {
            this.num = num;
            this.name = "线程" + num;
        }

        @Override
        public Integer get()
        {
            int i = 0;
            for (; i <= num; i++)
            {
                System.out.println(name + ": " + i);
                try
                {
                    TimeUnit.SECONDS.sleep(1);
                } catch (InterruptedException e)
                {
                    e.printStackTrace();
                }
            }
            return i;
        }

        @Override
        public String getName()
        {
            return name;
        }
    }
}
